package com.izhaowo.recommend.util;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import static java.util.Objects.nonNull;

/**
 * Created by HASEE on 2019/7/18.
 */
public class HBaseUtil {

    private static final Logger logger = LoggerFactory.getLogger(HBaseUtil.class);

    private static Configuration conf;
    private static Connection conn;

    private static final HBaseUtil INSTANCE = new HBaseUtil();

    private HBaseUtil() {
        try {
            if (conf == null) {
                conf = HBaseConfiguration.create();
                conf.set("hbase.zookeeper.quorum", "47.104.132.121");
                conf.set("zookeeper.znode.parent", "/hbase");
                conf.set("hbase.client.scanner.timeout.period", "60000000");
            }
        } catch (Exception e) {
            logger.error("HBase Configuration Initialization failure !");
            throw new RuntimeException(e);
        }
    }

    public static void init(String zkHost) {
        try {
            if (conf == null) {
                conf = HBaseConfiguration.create();
//                conf.set("hbase.zookeeper.property.clientPort", ConfigUtil.getInstance().getConfigVal("zkport", ConstantProperties.COMMON_PROP));
                conf.set("hbase.zookeeper.quorum", zkHost);
                conf.set("zookeeper.znode.parent", "/hbase");
//                conf.set("hbase.master", zkHost + ":60000");
                conf.set("hbase.client.scanner.timeout.period", "60000000");
//                conf.set("hbase.client.scanner.timeout.period", "20000");
            }
        } catch (Exception e) {
            logger.error("HBase Configuration Initialization failure !");
            throw new RuntimeException(e);
        }
    }

    /**
     * HBase连接
     */
    private Connection getConn() {
        if (conn == null || conn.isClosed()) {
            try {
                conn = ConnectionFactory.createConnection(conf);
                System.out.println("xxxxxx111" + conn.hashCode());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return conn;
    }

    /**
     * 获取HBase连接
     */
    public static Connection getConnection() {
        return INSTANCE.getConn();
    }

    /**
     * 获得链接
     *
     * @return
     */
    public static synchronized Connection getConnection2() {
        try {
            if (conn == null || conn.isClosed()) {
                conn = ConnectionFactory.createConnection(conf);
            }
            System.out.println("xxxxxx222" + conn.hashCode());
        } catch (IOException e) {
            logger.error("HBase 建立链接失败 ", e);
        }
        return conn;

    }

    /**
     * 获取  Table
     *
     * @param tableName 表名
     * @return
     * @throws IOException
     */
    public static Table getTable(String tableName) {
        try {
            return getConnection().getTable(TableName.valueOf(tableName));
        } catch (Exception e) {
            logger.error("Obtain Table failure !", e);
        }
        return null;
    }

    /**
     * 检索指定表的第一行记录。<br>
     * （如果在创建表时为此表指定了非默认的命名空间，则需拼写上命名空间名称，格式为【namespace:tablename】）。
     *
     * @param tableName  表名称(*)。
     * @param filterList 过滤器集合，可以为null。
     * @return
     */
    public static Result selectFirstResultRow(String tableName, FilterList filterList) {
        if (StringUtils.isBlank(tableName)) return null;
        Table table = null;
        try {
            table = getTable(tableName);
            Scan scan = new Scan();
            if (filterList != null) {
                scan.setFilter(filterList);
            }
            ResultScanner scanner = table.getScanner(scan);
            Iterator<Result> iterator = scanner.iterator();
            int index = 0;
            while (iterator.hasNext()) {
                Result rs = iterator.next();
                if (index == 0) {
                    scanner.close();
                    return rs;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    public static FilterList getFilterList(String status, String vocation, String vocation_id) {
        FilterList filter = null;
        if (StringUtils.isNotEmpty(status) || StringUtils.isNotEmpty(vocation)) {
            filter = new FilterList();
            if (StringUtils.isNotEmpty(status)) {
                Filter statusFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("status"),
                        CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(status)));
                filter.addFilter(statusFilter);

            }
            if (StringUtils.isNotEmpty(vocation)) {
                Filter vocationFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("vocation"),
                        CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(vocation)));
                filter.addFilter(vocationFilter);
            }
            if (StringUtils.isNotEmpty(vocation_id)) {
                Filter vocationFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("vocation_id"),
                        CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(vocation_id)));
                filter.addFilter(vocationFilter);
            }
        }


        return filter;
    }


    public static FilterList getTalkOrderFilter(String section, String city) {
        FilterList filter = null;
        if (StringUtils.isNotEmpty(section) || StringUtils.isNotEmpty(city)) {
            filter = new FilterList();
            Filter statusFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("section"),
                    CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(section)));
            filter.addFilter(statusFilter);
            Filter cityFilter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("city"),
                    CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(city)));
            filter.addFilter(cityFilter);
        }
        return filter;
    }


    public static ResultScanner queryRowsWithKeys(String tableName, String startRowkey, String endRowkey) {
        if (StringUtils.isBlank(tableName)) return null;
        Table table = null;
        try {
            table = getTable(tableName);
            Scan scan = new Scan();
            if (startRowkey != null) {
                scan.setStartRow(Bytes.toBytes(startRowkey));
            }
            if (endRowkey != null) {
                scan.setStopRow(Bytes.toBytes(endRowkey));
            }
            return table.getScanner(scan);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    /**
     * 异步往指定表添加数据
     *
     * @param tablename 表名
     * @param puts      需要添加的数据
     * @return long                返回执行时间
     * @throws IOException
     */
    public static long put(String tablename, List<Put> puts) throws Exception {
        long currentTime = System.currentTimeMillis();
        Connection conn = getConnection();
        final BufferedMutator.ExceptionListener listener = new BufferedMutator.ExceptionListener() {
            @Override
            public void onException(RetriesExhaustedWithDetailsException e, BufferedMutator mutator) {
                for (int i = 0; i < e.getNumExceptions(); i++) {
                    System.out.println("Failed to sent put " + e.getRow(i) + ".");
                    logger.error("Failed to sent put " + e.getRow(i) + ".");
                }
            }
        };
        BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tablename))
                .listener(listener);
        params.writeBufferSize(5 * 1024 * 1024);

        final BufferedMutator mutator = conn.getBufferedMutator(params);
        try {
            mutator.mutate(puts);
            mutator.flush();
        } finally {
            mutator.close();
//            closeConnect(conn);
        }
        return System.currentTimeMillis() - currentTime;
    }

    /**
     * 异步往指定表添加数据
     *
     * @param tablename 表名
     * @param put       需要添加的数据
     * @return long                返回执行时间
     * @throws IOException
     */
    public static long put(String tablename, Put put) throws Exception {
        return put(tablename, Arrays.asList(put));
    }

    /**
     * 往指定表添加数据
     *
     * @param tablename 表名
     * @param puts      需要添加的数据
     * @return long                返回执行时间
     * @throws IOException
     */
    public static long putByHTable(String tablename, List<?> puts) throws Exception {
        long currentTime = System.currentTimeMillis();
        Connection conn = getConnection();
        HTable htable = (HTable) conn.getTable(TableName.valueOf(tablename));
        htable.setAutoFlushTo(false);
        htable.setWriteBufferSize(5 * 1024 * 1024);
        try {
            htable.put((List<Put>) puts);
            htable.flushCommits();
        } finally {
            htable.close();
            closeConnect(conn);
        }
        return System.currentTimeMillis() - currentTime;
    }


    /**
     * 关闭连接
     *
     * @throws IOException
     */
    public static void closeConnect(Connection conn) {
        if (null != conn) {
            try {
                conn.close();
            } catch (Exception e) {
                logger.error("closeConnect failure !", e);
            }
        }
    }

    /**
     * 获取单条数据
     *
     * @param tablename
     * @param row
     * @return
     * @throws IOException
     */
    public static Result getRow(String tablename, byte[] row) {
        Table table = getTable(tablename);
        Result rs = null;
        if (table != null) {
            try {
                Get g = new Get(row);
                rs = table.get(g);
            } catch (IOException e) {
                logger.error("getRow failure !", e);
            } finally {
                try {
                    table.close();
                } catch (IOException e) {
                    logger.error("getRow failure !", e);
                }
            }
        }
        return rs;
    }

    /**
     * 创建表
     *
     * @param tableName
     * @throws Exception
     */
    public static void createTable(String tableName, String[] columnFamilies, boolean preBuildRegion) throws Exception {
        if (preBuildRegion) {
            String[] s = new String[]{"1", "2", "3", "4", "5", "6", "7", "8", "9", "A", "B", "C", "D", "E", "F"};
            int partition = 16;
            byte[][] splitKeys = new byte[partition - 1][];
            for (int i = 1; i < partition; i++) {
                splitKeys[i - 1] = Bytes.toBytes(s[i - 1]);
            }
            createTable(tableName, columnFamilies, splitKeys);
        } else {
            createTable(tableName, columnFamilies);
        }
    }

    /**
     * 建表
     *
     * @param tableName
     * @param cfs
     * @throws IOException
     */
    private static void createTable(String tableName, String[] cfs, byte[][] splitKeys) throws Exception {
        Connection conn = getConnection();
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        try {
            if (admin.tableExists(tableName)) {
                logger.warn("Table: {} is exists!", tableName);
                return;
            }
            HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
            for (int i = 0; i < cfs.length; i++) {
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cfs[i]);
                hColumnDescriptor.setCompressionType(Compression.Algorithm.SNAPPY);
                hColumnDescriptor.setMaxVersions(1);
                tableDesc.addFamily(hColumnDescriptor);
            }
            admin.createTable(tableDesc, splitKeys);
            logger.info("Table: {} create success!", tableName);
        } finally {
            admin.close();
            closeConnect(conn);
        }
    }

    /**
     * 建表
     *
     * @param tableName
     * @param cfs
     * @throws IOException
     */
    private static void createTable(String tableName, String[] cfs) throws Exception {
        Connection conn = getConnection();
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        try {
            if (admin.tableExists(tableName)) {
                logger.warn("Table: {} is exists!", tableName);
                return;
            }
            HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
            for (int i = 0; i < cfs.length; i++) {
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cfs[i]);
                hColumnDescriptor.setCompressionType(Compression.Algorithm.SNAPPY);
                hColumnDescriptor.setMaxVersions(1);
                tableDesc.addFamily(hColumnDescriptor);
            }
            admin.createTable(tableDesc);
            logger.info("Table: {} create success!", tableName);
        } finally {
            admin.close();
            closeConnect(conn);
        }
    }


    /**
     * 获取多行数据
     *
     * @param tablename
     * @param rows
     * @return
     * @throws Exception
     */
    public static <T> Result[] getRows(String tablename, List<T> rows) {
        Table table = getTable(tablename);
        List<Get> gets = null;
        Result[] results = null;
        try {
            if (table != null) {
                gets = new ArrayList<Get>();
                for (T row : rows) {
                    if (row != null) {
                        gets.add(new Get(Bytes.toBytes(String.valueOf(row))));
                    } else {
                        throw new RuntimeException("hbase have no data");
                    }
                }
            }
            if (gets.size() > 0) {
                results = table.get(gets);
            }
        } catch (IOException e) {
            logger.error("getRows failure !", e);
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                logger.error("table.close() failure !", e);
            }
        }
        return results;
    }

    /**
     * 扫描整张表，注意使用完要释放。
     *
     * @param tablename
     * @return
     * @throws IOException
     */
    public static ResultScanner get(String tablename, Filter filter) {
        Table table = getTable(tablename);
        ResultScanner results = null;
        if (table != null) {
            try {
                Scan scan = new Scan();
                if (null != filter)
                    scan.setFilter(filter);
                scan.setCaching(1000);
                results = table.getScanner(scan);
            } catch (IOException e) {
                logger.error("getResultScanner failure !", e);
            } finally {
                try {
                    table.close();
                } catch (IOException e) {
                    logger.error("table.close() failure !", e);
                }
            }
        }
        return results;
    }

    /**
     * 格式化输出结果
     */
    public static void formatRow(KeyValue[] rs) {
        for (KeyValue kv : rs) {
            System.out.println(" column family  :  " + Bytes.toString(kv.getFamily()));
            System.out.println(" column   :  " + Bytes.toString(kv.getQualifier()));
            System.out.println(" value   :  " + Bytes.toString(kv.getValue()));
            System.out.println(" timestamp   :  " + String.valueOf(kv.getTimestamp()));
            System.out.println("--------------------");
        }
    }

    /**
     * byte[] 类型的长整形数字转换成 long 类型
     *
     * @param byteNum
     * @return
     */
    public static long bytes2Long(byte[] byteNum) {
        long num = 0;
        for (int ix = 0; ix < 8; ++ix) {
            num <<= 8;
            num |= (byteNum[ix] & 0xff);
        }
        return num;
    }


    /**
     * 查询HBase表指定值
     *
     * @param tableName 表名
     * @param rowKey    主键
     * @param cf        列族
     * @param column    列名
     * @return 查询结果
     */
    public static String getValue(String tableName, String rowKey, String cf, String column) {
        String result1 = "";
        try (Table table = getTable(tableName)) {
            Get g = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(g);
            byte[] value = result.getValue(Bytes.toBytes(cf), Bytes.toBytes(column));
            result1 = Bytes.toString(value);
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        return result1;
    }

    /**
     * 单条查询HBase表
     *
     * @param tableName 表名
     * @param rowkey    主键
     * @return 查询结果
     */
    public static Result getRow(String tableName, String rowkey) {
        try (Table table = getTable(tableName)) {
            Get get = new Get(Bytes.toBytes(rowkey));
            return table.get(get);
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        return null;
    }

    /**
     * 全表扫描
     *
     * @param tableName 表名
     * @return 查询结果（ResultScanner）
     */
    public static ResultScanner getScanner(String tableName) {
        try (Table table = getTable(tableName)) {
            Scan scan = new Scan();
            scan.setCaching(1000);
            return table.getScanner(scan);
        } catch (IOException ioe) {
            ioe.printStackTrace();
        }
        return null;
    }
}
